﻿using System;
using System.Collections.Generic;
using System.Data;
using System.Data.SqlClient;
using System.IO;
using System.Linq;
using System.Reflection;
using System.Text;
using System.Xml;
using CoreLab.Oracle;
using Common.Utilities;
using Common.Utilities.DLR;

namespace Workflows.Components.Entities.Providers
{
    [Serializable]
    public class EntityProvider
    {
        #region props
        private SourceType _Source;
        /// <summary></summary>
        public SourceType Source
        {
            get
            {
                return this._Source;
            }
            set
            {
                this._Source = value;
            }
        }

        private string _FilePath;
        /// <summary></summary>
        public string FilePath
        {
            get
            {
                return this._FilePath;
            }
            set
            {
                this._FilePath = value;
            }
        }

        private DataProviderType _DbProvider;
        /// <summary></summary>
        public DataProviderType DbProvider
        {
            get
            {
                return this._DbProvider;
            }
            set
            {
                this._DbProvider = value;
            }
        }

        private string _ConnectionString;
        /// <summary></summary>
        public string ConnectionString
        {
            get
            {
                return this._ConnectionString;
            }
            set
            {
                this._ConnectionString = value;
            }
        }

        private string _SelectSql;
        /// <summary></summary>
        public string SelectSql
        {
            get
            {
                return this._SelectSql;
            }
            set
            {
                this._SelectSql = value;
            }
        }

        private string _IdAttributeName;
        /// <summary></summary>
        public string IdAttributeName
        {
            get
            {
                return this._IdAttributeName;
            }
            set
            {
                this._IdAttributeName = value;
            }
        }

        private string _EntityCategory;
        /// <summary></summary>
        public string EntityCategory
        {
            get
            {
                return this._EntityCategory;
            }
            set
            {
                this._EntityCategory = value;
            }
        }

        public EntityMetaData MetaData
        {
            get { return _MetaData; }
            set { _MetaData = value; }
        }

        private EntityMetaData _MetaData;
        #endregion

        #region ctor
        public EntityProvider(string filePath, string idAttrName, string entityCategory)
        {
            this._Source = SourceType.File;
            this._FilePath = filePath;
            this._DbProvider = DataProviderType.File;
            this._ConnectionString = string.Empty;
            this._SelectSql = string.Empty;

            this._IdAttributeName = idAttrName;
            this._EntityCategory = entityCategory;

            if (!string.IsNullOrEmpty(filePath) && File.Exists(filePath))
                this.MetaData = this.ReadMetaDataFromFile();
            else
                this.MetaData = new EntityMetaData(idAttrName, entityCategory, new List<AttributeMetaData>());
        }

        public EntityProvider(DataProviderType dbprovider, string connStr, string sql,
            string idAttrName, string entityCategory)
        {
            this._Source = SourceType.Database;
            this._FilePath = string.Empty;
            this._DbProvider = dbprovider;
            this._ConnectionString = connStr;
            this._SelectSql = sql;

            this._IdAttributeName = idAttrName;
            this._EntityCategory = entityCategory;

            if (!string.IsNullOrEmpty(connStr) && !string.IsNullOrEmpty(sql))
                this.MetaData = this.ReadMetaDataFromDB();
            else
                this.MetaData = new EntityMetaData(idAttrName, entityCategory, new List<AttributeMetaData>());
        }
        #endregion

        #region meta data
        private EntityMetaData ReadMetaDataFromFile()
        {
            List<AttributeMetaData> attributeMetaDatas = new List<AttributeMetaData>();
            XmlDocument xDoc0 = new XmlDocument();
            xDoc0.Load(this._FilePath);
            XmlNode rootNode = xDoc0.DocumentElement;

            string metaDataFilePath = XmlDataUtil.GetAttributeValue(rootNode, "MetaDataFilePath", "");
            if (!string.IsNullOrEmpty(metaDataFilePath) && File.Exists(metaDataFilePath))
            {
                XmlDocument xDoc = new XmlDocument();
                xDoc.Load(metaDataFilePath);
                XmlNode entityNode = xDoc.SelectSingleNode("EntityMetaData/Entity[@IsRoot=\"" + true.ToString() + "\"]");
                XmlNodeList attrNodes = entityNode.SelectNodes("Attribute");
                foreach (XmlNode attrNode in attrNodes)
                {
                    string attrName = attrNode.Attributes.GetNamedItem("AttributeName").Value;
                    AttributeType attrType =
                        (AttributeType)int.Parse(attrNode.Attributes.GetNamedItem("AttributeType").Value);
                    string comparerAssemblyFileName = attrNode.Attributes.GetNamedItem("ComparerAssemblyName").Value;
                    string comparerTypeFullName = attrNode.Attributes.GetNamedItem("ComparerTypeName").Value;
                    double posW = double.Parse(attrNode.Attributes.GetNamedItem("PositiveMatchWeight").Value);
                    double negW = double.Parse(attrNode.Attributes.GetNamedItem("NegativeMatchWeight").Value);
                    bool isExternal = bool.Parse(attrNode.Attributes.GetNamedItem("IsExternal").Value);
                    DataProviderType exDbProvider =
                        (DataProviderType) int.Parse(attrNode.Attributes.GetNamedItem("ExDbProvider").Value);
                    string exConnStr = attrNode.Attributes.GetNamedItem("ExConnStr").Value;
                    string exSql = attrNode.Attributes.GetNamedItem("ExSql").Value;
                    Dictionary<string, string> comparerSettings = new Dictionary<string, string>();
                    XmlNodeList settingNodes = attrNode.SelectNodes("Setting");
                    if (settingNodes != null && settingNodes.Count > 0)
                    {
                        foreach (XmlNode settingNode in settingNodes)
                        {
                            string name = XmlDataUtil.GetAttributeValue(settingNode, "Name", "");
                            string value = XmlDataUtil.GetAttributeValue(settingNode, "Value", "");
                            comparerSettings.Add(name, value);
                        }
                    }

                    AttributeMetaData attrMetaData =
                        new AttributeMetaData(attrName, attrType,
                                              comparerAssemblyFileName,
                                              comparerTypeFullName,
                                              comparerSettings, posW, negW,
                                              isExternal, exDbProvider, exConnStr, exSql);
                    attributeMetaDatas.Add(attrMetaData);
                }
            }

            return new EntityMetaData(this._IdAttributeName, this._EntityCategory, attributeMetaDatas);
        }

        private EntityMetaData ReadMetaDataFromDB()
        {
            TableColumn[] cols = DbUtil.GetTableColumns(this._DbProvider, this._ConnectionString, this._SelectSql);
            if (cols != null && cols.Length > 0)
            {
                List<AttributeMetaData> attributeMetaDatas = new List<AttributeMetaData>();
                foreach (TableColumn col in cols)
                {
                    AttributeMetaData attrMetaData =
                        new AttributeMetaData(col.ColumnName,
                                              AttributeType.NativeAttribute,
                                              string.Empty, string.Empty, 
                                              new Dictionary<string, string>(),  1.0, -1.0,
                                              false, DataProviderType.MSSQL, string.Empty,string.Empty);
                    attributeMetaDatas.Add(attrMetaData);
                }

                return new EntityMetaData(this._IdAttributeName, this._EntityCategory, attributeMetaDatas);
            }
            else
                return new EntityMetaData(this._IdAttributeName, this._EntityCategory, new List<AttributeMetaData>());
        }
        #endregion

        #region read all
        public Dictionary<string, Entity> ReadEntities()
        {
            Dictionary<string, Entity> entities = new Dictionary<string, Entity>();
            List<Entity> entityList = new List<Entity>();
            if (this._Source == SourceType.Database)
                entityList= this.ReadEntitiesFromDb(this._MetaData);
            else
                entityList = this.ReadEntitiesFromFile(this._MetaData);
            foreach(Entity entity in entityList)
            {
                entities.Add(entity.Id, entity);
            }
            return entities;
        }

        private List<Entity> ReadEntitiesFromDb(EntityMetaData metaData)
        {
            List<Entity> entities = new List<Entity>();
            IDbConnection conn = null;
            if (this._DbProvider == DataProviderType.MSSQL)
                conn = new SqlConnection(this._ConnectionString);
            else
                conn = new OracleConnection(this._ConnectionString);
            try
            {
                conn.Open();
                IDbCommand cmd = conn.CreateCommand();
                cmd.CommandType = CommandType.Text;
                cmd.CommandText = this._SelectSql;
                IDbDataAdapter da = null;
                if (this._DbProvider == DataProviderType.MSSQL)
                    da = new SqlDataAdapter((SqlCommand)cmd);
                else
                    da = new OracleDataAdapter((OracleCommand) cmd);
                DataSet ds = new DataSet();
                da.Fill(ds);
                DataTable dt = ds.Tables[0];

                foreach(DataRow dr in dt.Rows)
                {
                    string id = (string)dr[metaData.IdAttributeName];
                    Dictionary<string, object> attrValues = new Dictionary<string, object>();
                    foreach(AttributeMetaData attrMetaData in metaData.Attributes)
                    {
                        if(dt.Columns.Contains(attrMetaData.AttributeName) && 
                            dr[attrMetaData.AttributeName]!=null)
                        {
                            attrValues.Add(attrMetaData.AttributeName, dr[attrMetaData.AttributeName]);
                        }
                        else
                        {
                            attrValues.Add(attrMetaData.AttributeName, null);
                        }
                    }
                    Entity entity = new Entity(id,attrValues);
                    entities.Add(entity);
                }
            }
            catch(Exception ex)
            {
                throw new Exception("Unable to read entities from db: "+ex.ToString());
            }
            finally
            {
                conn.Close();
            }

            return entities;
        }

        private List<Entity> ReadEntitiesFromFile(EntityMetaData metaData)
        {
            List<Entity> entities = new List<Entity>();

            return entities;
        }

        public List<int> GetEntityIDs()
        {
            if (this.Source == SourceType.Database)
                return this.GetEntityIDsFromDb(this.MetaData);
            else
                return this.GetEntityIDsFromFile(this.MetaData);
        }

        private List<int> GetEntityIDsFromDb(EntityMetaData metaData)
        {
            List<int> entityIDs = new List<int>();
            IDbConnection conn = null;
            if (this._DbProvider == DataProviderType.MSSQL)
                conn = new SqlConnection(this._ConnectionString);
            else
                conn = new OracleConnection(this._ConnectionString);
            try
            {
                conn.Open();
                IDbCommand cmd = conn.CreateCommand();
                cmd.CommandType = CommandType.Text;
                cmd.CommandText = string.Format("select {0} from ({1}) temp",
                                                this.MetaData.IdAttributeName, this._SelectSql);
                IDbDataAdapter da = null;
                if (this._DbProvider == DataProviderType.MSSQL)
                    da = new SqlDataAdapter((SqlCommand)cmd);
                else
                    da = new OracleDataAdapter((OracleCommand)cmd);
                DataSet ds = new DataSet();
                da.Fill(ds);
                DataTable dt = ds.Tables[0];

                foreach (DataRow dr in dt.Rows)
                {
                    int id = (int)dr[metaData.IdAttributeName];
                    entityIDs.Add(id);
                }
            }
            catch (Exception ex)
            {
                throw new Exception("Unable to read entities from db: " + ex.ToString());
            }
            finally
            {
                conn.Close();
            }
            return entityIDs;

        }

        private List<int> GetEntityIDsFromFile(EntityMetaData metaData)
        {

            List<int> entityIDs = new List<int>();

            return entityIDs;
        }
        #endregion

        #region get by pk/fk
        public Entity GetEntityByPK(string pk)
        {
            Entity entity = null;
            IDbConnection conn = null;
            if (this._Source == SourceType.File)
                throw new NotImplementedException("Not implemented");

            if (this._DbProvider == DataProviderType.MSSQL)
                conn = new SqlConnection(this._ConnectionString);
            else
                conn = new OracleConnection(this._ConnectionString);
            try
            {
                conn.Open();
                IDbCommand cmd = conn.CreateCommand();
                string sql = string.Format("select * from ({0}) as tmp where tmp.{1}={2}",
                                           this.SelectSql, this.IdAttributeName, pk);
                cmd.CommandType = CommandType.Text;
                cmd.CommandText = sql;
                IDataReader reader = cmd.ExecuteReader();
                if(reader.Read())
                {
                    Dictionary<string, object> attrValues = new Dictionary<string, object>();
                    foreach (AttributeMetaData attrMetaData in this._MetaData.Attributes)
                    {
                        if (reader[attrMetaData.AttributeName] != null)
                        {
                            attrValues.Add(attrMetaData.AttributeName, reader[attrMetaData.AttributeName]);
                        }
                        else
                        {
                            attrValues.Add(attrMetaData.AttributeName, null);
                        }
                    }
                    entity = new Entity(pk, attrValues);
                }
                reader.Close();
            }
            catch (Exception ex)
            {
                throw new Exception("Unable to read entities from db: " + ex.ToString());
            }
            finally
            {
                conn.Close();
            }
            return entity;
        }

        public Dictionary<string, Entity> GetEntitiesByPK(List<string> pks)
        {
            Dictionary<string, Entity> entities = new Dictionary<string, Entity>();
            IDbConnection conn = null;
            

            if (this._DbProvider == DataProviderType.MSSQL)
                conn = new SqlConnection(this._ConnectionString);
            else
                conn = new OracleConnection(this._ConnectionString);
            try
            {
                conn.Open();
                IDbCommand cmd = conn.CreateCommand();
                foreach (string pk in pks)
                {
                    string sql = string.Format("select * from ({0}) as tmp where tmp.{1}={2}",
                                           this.SelectSql, this.IdAttributeName, pk);
                    cmd.CommandType = CommandType.Text;
                    cmd.CommandText = sql;
                    IDataReader reader = cmd.ExecuteReader();
                    if (reader.Read())
                    {
                        Entity entity = null;
                        Dictionary<string, object> attrValues = new Dictionary<string, object>();
                        foreach (AttributeMetaData attrMetaData in this._MetaData.Attributes)
                        {
                            if (reader[attrMetaData.AttributeName] != null)
                            {
                                attrValues.Add(attrMetaData.AttributeName, reader[attrMetaData.AttributeName]);
                            }
                            else
                            {
                                attrValues.Add(attrMetaData.AttributeName, null);
                            }
                        }
                        entity = new Entity(pk, attrValues);

                        entities.Add(pk, entity);
                    }
                    reader.Close();
                }
            }
            catch (Exception ex)
            {
                throw new Exception("Unable to read entities from db: " + ex.ToString());
            }
            finally
            {
                conn.Close();
            }

            return entities;
        }

        public List<Entity> GetEntitiesByFK(string fkAttrName, int fk)
        {
            if (this._Source == SourceType.File)
                throw new NotImplementedException("Not implemented");

            List<Entity> entities = new List<Entity>();
            IDbConnection conn = null;
            if (this._Source == SourceType.File)
                throw new NotImplementedException("Not implemented");

            if (this._DbProvider == DataProviderType.MSSQL)
                conn = new SqlConnection(this._ConnectionString);
            else
                conn = new OracleConnection(this._ConnectionString);
            try
            {
                conn.Open();
                IDbCommand cmd = conn.CreateCommand();
                string sql = string.Format("select * from ({0}) as tmp where tmp.{1}={2}",
                                           this.SelectSql, fkAttrName, fk);
                cmd.CommandType = CommandType.Text;
                cmd.CommandText = sql;
                IDataReader reader = cmd.ExecuteReader();
                while (reader.Read())
                {
                    string pk = (string)reader[this._MetaData.IdAttributeName];
                    Dictionary<string, object> attrValues = new Dictionary<string, object>();
                    foreach (AttributeMetaData attrMetaData in this._MetaData.Attributes)
                    {
                        if (reader[attrMetaData.AttributeName] != null)
                        {
                            attrValues.Add(attrMetaData.AttributeName, reader[attrMetaData.AttributeName]);
                        }
                        else
                        {
                            attrValues.Add(attrMetaData.AttributeName, null);
                        }
                    }
                    Entity entity = new Entity(pk, attrValues);
                    entities.Add(entity);
                }
                reader.Close();
            }
            catch (Exception ex)
            {
                throw new Exception("Unable to read entities from db: " + ex.ToString());
            }
            finally
            {
                conn.Close();
            }

            return entities;
        }

        public List<string> GetEntityIDsByFK(string fkAttrName, string fk)
        {
            if (this._Source == SourceType.File)
                throw new NotImplementedException("Not implemented");

            List<string> entityIDs = new List<string>();
            IDbConnection conn = null;
            if (this._Source == SourceType.File)
                throw new NotImplementedException("Not implemented");

            if (this._DbProvider == DataProviderType.MSSQL)
                conn = new SqlConnection(this._ConnectionString);
            else
                conn = new OracleConnection(this._ConnectionString);
            try
            {
                conn.Open();
                IDbCommand cmd = conn.CreateCommand();
                string sql = string.Format("select * from ({0}) as tmp where tmp.{1}={2}",
                                           this.SelectSql, fkAttrName, fk);
                cmd.CommandType = CommandType.Text;
                cmd.CommandText = sql;
                IDataReader reader = cmd.ExecuteReader();
                while (reader.Read())
                {
                    string pk = (string)reader[this._MetaData.IdAttributeName];
                    entityIDs.Add(pk);
                }
                reader.Close();
            }
            catch (Exception ex)
            {
                throw new Exception("Unable to read entities from db: " + ex.ToString());
            }
            finally
            {
                conn.Close();
            }

            return entityIDs;
        }
        #endregion

        #region partition
        public Dictionary<int,List<Entity>> GetPartitionedEntities(
            string partitionAttrName, int entityCountPerPartition)
        {
            if (this._Source == SourceType.File)
                throw new NotImplementedException("Not implemented");

            Dictionary<int, List<Entity>> partitionedEntities = new Dictionary<int, List<Entity>>();
            IDbConnection conn = null;
            if (this._Source == SourceType.File)
                throw new NotImplementedException("Not implemented");

            if (this._DbProvider == DataProviderType.MSSQL)
                conn = new SqlConnection(this._ConnectionString);
            else
                conn = new OracleConnection(this._ConnectionString);
            try
            {
                conn.Open();
                IDbCommand cmd = conn.CreateCommand();
                string sql = string.Format("select * from ({0}) as tmp order by tmp.{1}",
                                           this.SelectSql, partitionAttrName);
                cmd.CommandType = CommandType.Text;
                cmd.CommandText = sql;
                IDataReader reader = cmd.ExecuteReader();
                List<Entity> currentPartitionedEntities = new List<Entity>();
                while (reader.Read())
                {
                    string pk = (string)reader[this._MetaData.IdAttributeName];
                    Dictionary<string, object> attrValues = new Dictionary<string, object>();
                    foreach (AttributeMetaData attrMetaData in this._MetaData.Attributes)
                    {
                        if (reader[attrMetaData.AttributeName] != null)
                        {
                            attrValues.Add(attrMetaData.AttributeName, reader[attrMetaData.AttributeName]);
                        }
                        else
                        {
                            attrValues.Add(attrMetaData.AttributeName, null);
                        }
                    }
                    Entity entity = new Entity(pk, attrValues);
                    currentPartitionedEntities.Add(entity);
                    if(currentPartitionedEntities.Count>=entityCountPerPartition)
                    {
                        int partitionID = partitionedEntities.Count + 1;
                        partitionedEntities.Add(partitionID, currentPartitionedEntities);
                        currentPartitionedEntities = new List<Entity>();
                    }
                }
                reader.Close();
                if(currentPartitionedEntities.Count>0)
                {
                    int partitionID = partitionedEntities.Count + 1;
                    partitionedEntities.Add(partitionID, currentPartitionedEntities);
                }
            }
            catch (Exception ex)
            {
                throw new Exception("Unable to read entities from db: " + ex.ToString());
            }
            finally
            {
                conn.Close();
            }

            return partitionedEntities;
        }

        public Dictionary<int, List<Entity>> GetPartitionedEntities(
            string partitionAttrName, List<Range<int>> attrValueRanges)
        {
            if (this._Source == SourceType.File)
                throw new NotImplementedException("Not implemented");

            List<int> fromValues = new List<int>();
            Dictionary<int, Range<int>> ranges = new Dictionary<int, Range<int>>();
            foreach(Range<int> range in attrValueRanges)
            {
                if (!fromValues.Contains(range.FromValue))
                {
                    fromValues.Add(range.FromValue);
                    ranges.Add(range.FromValue, range);
                }
            }
            fromValues.Sort();

            Dictionary<int, List<Entity>> partitionedEntities = new Dictionary<int, List<Entity>>();
            IDbConnection conn = null;
            if (this._Source == SourceType.File)
                throw new NotImplementedException("Not implemented");

            if (this._DbProvider == DataProviderType.MSSQL)
                conn = new SqlConnection(this._ConnectionString);
            else
                conn = new OracleConnection(this._ConnectionString);
            try
            {
                conn.Open();
                IDbCommand cmd = conn.CreateCommand();
                string sql = string.Format("select * from ({0}) as tmp order by tmp.{1}",
                                           this.SelectSql, partitionAttrName);
                cmd.CommandType = CommandType.Text;
                cmd.CommandText = sql;
                IDataReader reader = cmd.ExecuteReader();
                List<Entity> currentPartitionedEntities = new List<Entity>();
                int currentRangeIndex = 0;
                while (reader.Read())
                {
                    string pk = (string)reader[this._MetaData.IdAttributeName];
                    double attrValue = double.Parse(reader[partitionAttrName].ToString());
                    Dictionary<string, object> attrValues = new Dictionary<string, object>();
                    foreach (AttributeMetaData attrMetaData in this._MetaData.Attributes)
                    {
                        if (reader[attrMetaData.AttributeName] != null)
                        {
                            attrValues.Add(attrMetaData.AttributeName, reader[attrMetaData.AttributeName]);
                        }
                        else
                        {
                            attrValues.Add(attrMetaData.AttributeName, null);
                        }
                    }
                    Entity entity = new Entity(pk, attrValues);
                    int currentLowValue = fromValues[currentRangeIndex];
                    Range<int> currentRange = ranges[currentLowValue];
                    if (attrValue <= currentRange.ToValue)
                    {
                        currentPartitionedEntities.Add(entity);
                    }
                    else
                    {
                        int partitionID = partitionedEntities.Count + 1;
                        partitionedEntities.Add(partitionID, currentPartitionedEntities);
                        currentPartitionedEntities = new List<Entity>();

                        while (attrValue > currentRange.ToValue)
                        {
                            if (currentRangeIndex < ranges.Count - 1)
                            {
                                currentRangeIndex++;
                                currentLowValue = fromValues[currentRangeIndex];
                                currentRange = ranges[currentLowValue];
                            }
                        }
                        if (currentRangeIndex < ranges.Count)
                        {
                            currentPartitionedEntities.Add(entity);
                        }
                        else
                            break;
                    }
                }
                reader.Close();
                if (currentPartitionedEntities.Count > 0)
                {
                    int partitionID = partitionedEntities.Count + 1;
                    partitionedEntities.Add(partitionID, currentPartitionedEntities);
                }
            }
            catch (Exception ex)
            {
                throw new Exception("Unable to read entities from db: " + ex.ToString());
            }
            finally
            {
                conn.Close();
            }

            return partitionedEntities;
        }

        public Dictionary<int,List<int>> GetPartitionedEntityIDs(
            string partitionAttrName, int entityCountPerPartition)
        {
            if (this._Source == SourceType.File)
                throw new NotImplementedException("Not implemented");

            Dictionary<int, List<int>> partitionedEntityIDs = new Dictionary<int, List<int>>();
            IDbConnection conn = null;
            if (this._Source == SourceType.File)
                throw new NotImplementedException("Not implemented");

            if (this._DbProvider == DataProviderType.MSSQL)
                conn = new SqlConnection(this._ConnectionString);
            else
                conn = new OracleConnection(this._ConnectionString);
            try
            {
                conn.Open();
                IDbCommand cmd = conn.CreateCommand();
                string sql = string.Format("select * from ({0}) as tmp order by tmp.{1}",
                                           this.SelectSql, partitionAttrName);
                cmd.CommandType = CommandType.Text;
                cmd.CommandText = sql;
                IDataReader reader = cmd.ExecuteReader();
                List<int> currentPartitionedEntityIDs = new List<int>();
                while (reader.Read())
                {
                    int pk = (int)reader[this._MetaData.IdAttributeName];
                    currentPartitionedEntityIDs.Add(pk);
                    if (currentPartitionedEntityIDs.Count >= entityCountPerPartition)
                    {
                        int partitionID = partitionedEntityIDs.Count + 1;
                        partitionedEntityIDs.Add(partitionID, currentPartitionedEntityIDs);
                        currentPartitionedEntityIDs = new List<int>();
                    }
                }
                reader.Close();
                if (currentPartitionedEntityIDs.Count > 0)
                {
                    int partitionID = partitionedEntityIDs.Count + 1;
                    partitionedEntityIDs.Add(partitionID, currentPartitionedEntityIDs);
                }
            }
            catch (Exception ex)
            {
                throw new Exception("Unable to read entities from db: " + ex.ToString());
            }
            finally
            {
                conn.Close();
            }

            return partitionedEntityIDs;
        }

        public Dictionary<int,List<int>> GetPartitionedEntityIDs(
            string partitionAttrName, List<Range<int>> attrValueRanges)
        {
            if (this._Source == SourceType.File)
                throw new NotImplementedException("Not implemented");


            List<int> fromValues = new List<int>();
            Dictionary<int, Range<int>> ranges = new Dictionary<int, Range<int>>();
            foreach (Range<int> range in attrValueRanges)
            {
                if (!fromValues.Contains(range.FromValue))
                {
                    fromValues.Add(range.FromValue);
                    ranges.Add(range.FromValue, range);
                }
            }
            fromValues.Sort();

            Dictionary<int, List<int>> partitionedEntityIDs = new Dictionary<int, List<int>>();
            IDbConnection conn = null;
            if (this._Source == SourceType.File)
                throw new NotImplementedException("Not implemented");

            if (this._DbProvider == DataProviderType.MSSQL)
                conn = new SqlConnection(this._ConnectionString);
            else
                conn = new OracleConnection(this._ConnectionString);
            try
            {
                conn.Open();
                IDbCommand cmd = conn.CreateCommand();
                string sql = string.Format("select * from ({0}) as tmp order by tmp.{1}",
                                           this.SelectSql, partitionAttrName);
                cmd.CommandType = CommandType.Text;
                cmd.CommandText = sql;
                IDataReader reader = cmd.ExecuteReader();
                List<int> currentPartitionedEntityIDs = new List<int>();
                int currentRangeIndex = 0;
                while (reader.Read())
                {
                    int pk = (int)reader[this._MetaData.IdAttributeName];
                    double attrValue = double.Parse(reader[partitionAttrName].ToString());
                    
                    int currentLowValue = fromValues[currentRangeIndex];
                    Range<int> currentRange = ranges[currentLowValue];
                    if (attrValue <= currentRange.ToValue)
                    {
                        currentPartitionedEntityIDs.Add(pk);
                    }
                    else
                    {
                        int partitionID = partitionedEntityIDs.Count + 1;
                        partitionedEntityIDs.Add(partitionID, currentPartitionedEntityIDs);
                        currentPartitionedEntityIDs = new List<int>();

                        while (attrValue > currentRange.ToValue)
                        {
                            if (currentRangeIndex < ranges.Count - 1)
                            {
                                currentRangeIndex++;
                                currentLowValue = fromValues[currentRangeIndex];
                                currentRange = ranges[currentLowValue];
                            }
                        }
                        if (currentRangeIndex < ranges.Count)
                        {
                            currentPartitionedEntityIDs.Add(pk);
                        }
                        else
                            break;
                    }
                }
                reader.Close();
                if (currentPartitionedEntityIDs.Count > 0)
                {
                    int partitionID = partitionedEntityIDs.Count + 1;
                    partitionedEntityIDs.Add(partitionID, currentPartitionedEntityIDs);
                }
            }
            catch (Exception ex)
            {
                throw new Exception("Unable to read entities from db: " + ex.ToString());
            }
            finally
            {
                conn.Close();
            }

            return partitionedEntityIDs;
        }

        public Dictionary<int,List<int>> GetPartitionedEntityIDs(
            List<int> srcEntityIDs, string partitionAttrName, int entityCountPerPartition)
        {
            if (this._Source == SourceType.File)
                throw new NotImplementedException("Not implemented");

            if (srcEntityIDs == null && srcEntityIDs.Count == 0)
                return null;

            Dictionary<int, List<int>> partitionedEntityIDs = new Dictionary<int, List<int>>();
            if (srcEntityIDs.Count <= entityCountPerPartition)
            {
                partitionedEntityIDs.Add(1, srcEntityIDs);
                return partitionedEntityIDs;
            }

            IDbConnection conn = null;
            if (this._Source == SourceType.File)
                throw new NotImplementedException("Not implemented");

            if (this._DbProvider == DataProviderType.MSSQL)
                conn = new SqlConnection(this._ConnectionString);
            else
                conn = new OracleConnection(this._ConnectionString);
            try
            {
                conn.Open();
                IDbCommand cmd = conn.CreateCommand();

                string tmpTableName = "tmp_Key_" + this._IdAttributeName;
                string keyFieldName = this._IdAttributeName;
                DbUtil.NewTempKeyTable(this._ConnectionString, tmpTableName, keyFieldName,
                                       srcEntityIDs);

                string sql = string.Format(
                    "select * from ({0}) as tmp where " +
                    "tmp.{1} in (select {2} from {3}) order by tmp.{4}",
                    this.SelectSql, this._IdAttributeName, keyFieldName,
                    tmpTableName, partitionAttrName);
                cmd.CommandType = CommandType.Text;
                cmd.CommandText = sql;
                IDataReader reader = cmd.ExecuteReader();
                List<int> currentPartitionedEntityIDs = new List<int>();
                while (reader.Read())
                {
                    int pk = (int)reader[this._MetaData.IdAttributeName];
                    //if (srcEntityIDs.Contains(pk))
                    currentPartitionedEntityIDs.Add(pk);
                    if (currentPartitionedEntityIDs.Count >= entityCountPerPartition)
                    {
                        int partitionID = partitionedEntityIDs.Count + 1;
                        partitionedEntityIDs.Add(partitionID, currentPartitionedEntityIDs);
                        currentPartitionedEntityIDs = new List<int>();
                    }
                }
                reader.Close();

                DbUtil.DropTempTable(this._DbProvider, this._ConnectionString, tmpTableName);
                if (currentPartitionedEntityIDs.Count > 0)
                {
                    int partitionID = partitionedEntityIDs.Count + 1;
                    partitionedEntityIDs.Add(partitionID, currentPartitionedEntityIDs);
                }
            }
            catch (Exception ex)
            {
                throw new Exception("Unable to read entities from db: " + ex.ToString());
            }
            finally
            {
                conn.Close();
            }

            return partitionedEntityIDs;
        }

        public Dictionary<int, List<int>> GetPartitionedEntityIDs(
            List<int> srcEntityIDs, string partitionAttrName, 
            int minCountPerPartition,
            List<Range<int>> attrValueRanges)
        {
            if (this._Source == SourceType.File)
                throw new NotImplementedException("Not implemented");

            if (srcEntityIDs == null && srcEntityIDs.Count == 0)
                return null;

            Dictionary<int, List<int>> partitionedEntityIDs = new Dictionary<int, List<int>>();
            if (srcEntityIDs.Count <= minCountPerPartition)
            {
                partitionedEntityIDs.Add(1, srcEntityIDs);
                return partitionedEntityIDs;
            }

            List<int> fromValues = new List<int>();
            Dictionary<int, Range<int>> ranges = new Dictionary<int, Range<int>>();
            foreach (Range<int> range in attrValueRanges)
            {
                if (!fromValues.Contains(range.FromValue))
                {
                    fromValues.Add(range.FromValue);
                    ranges.Add(range.FromValue, range);
                }
            }
            fromValues.Sort();

            IDbConnection conn = null;
            if (this._Source == SourceType.File)
                throw new NotImplementedException("Not implemented");

            if (this._DbProvider == DataProviderType.MSSQL)
                conn = new SqlConnection(this._ConnectionString);
            else
                conn = new OracleConnection(this._ConnectionString);
            try
            {
                conn.Open();
                IDbCommand cmd = conn.CreateCommand();

                string tmpTableName = "tmp_Key_" + this._IdAttributeName;
                string keyFieldName = this._IdAttributeName;
                DbUtil.NewTempKeyTable(this._ConnectionString, tmpTableName, keyFieldName,
                                       srcEntityIDs);

                string sql = string.Format(
                    "select * from ({0}) as tmp where " +
                    "tmp.{1} in (select {2} from {3}) order by tmp.{4}",
                    this.SelectSql, this._IdAttributeName, keyFieldName,
                    tmpTableName, partitionAttrName);
                cmd.CommandType = CommandType.Text;
                cmd.CommandText = sql;
                IDataReader reader = cmd.ExecuteReader();
                List<int> currentPartitionedEntityIDs = new List<int>();
                int currentRangeIndex = 0;
                while (reader.Read())
                {
                    int pk = (int)reader[this._MetaData.IdAttributeName];
                    double attrValue = double.Parse(reader[partitionAttrName].ToString());
                    
                    int currentLowValue = fromValues[currentRangeIndex];
                    Range<int> currentRange = ranges[currentLowValue];
                    if (attrValue <= currentRange.ToValue)
                    {
                        currentPartitionedEntityIDs.Add(pk);
                    }
                    else
                    {
                        int partitionID = partitionedEntityIDs.Count + 1;
                        partitionedEntityIDs.Add(partitionID, currentPartitionedEntityIDs);
                        currentPartitionedEntityIDs = new List<int>();

                        while (attrValue > currentRange.ToValue)
                        {
                            if (currentRangeIndex < ranges.Count - 1)
                            {
                                currentRangeIndex++;
                                currentLowValue = fromValues[currentRangeIndex];
                                currentRange = ranges[currentLowValue];
                            }
                        }
                        if (currentRangeIndex < ranges.Count)
                        {
                            currentPartitionedEntityIDs.Add(pk);
                        }
                        else
                            break;
                    }
                }
                reader.Close();

                DbUtil.DropTempTable(this._DbProvider, this._ConnectionString, tmpTableName);
                if (currentPartitionedEntityIDs.Count > 0)
                {
                    int partitionID = partitionedEntityIDs.Count + 1;
                    partitionedEntityIDs.Add(partitionID, currentPartitionedEntityIDs);
                }
            }
            catch (Exception ex)
            {
                throw new Exception("Unable to read entities from db: " + ex.ToString());
            }
            finally
            {
                conn.Close();
            }

            return partitionedEntityIDs;
        }

        public Dictionary<int,List<int>> GetPartitionedEntityIDsByDistinctAttrValue(
            List<int> srcEntityIDs, string partitionAttrName)
        {
            if (this._Source == SourceType.File)
                throw new NotImplementedException("Not implemented");

            if (srcEntityIDs == null && srcEntityIDs.Count == 0)
                return null;

            Dictionary<int, List<int>> partitionedEntityIDs = new Dictionary<int, List<int>>();

            IDbConnection conn = null;
            if (this._Source == SourceType.File)
                throw new NotImplementedException("Not implemented");

            if (this._DbProvider == DataProviderType.MSSQL)
                conn = new SqlConnection(this._ConnectionString);
            else
                conn = new OracleConnection(this._ConnectionString);
            try
            {
                conn.Open();
                IDbCommand cmd = conn.CreateCommand();

                string tmpTableName = "tmp_Key_" + this._IdAttributeName;
                string keyFieldName = this._IdAttributeName;
                DbUtil.NewTempKeyTable(this._ConnectionString, tmpTableName, keyFieldName,
                                       srcEntityIDs);

                string sql = string.Format(
                    "select * from ({0}) as tmp where " +
                    "tmp.{1} in (select {2} from {3}) order by tmp.{4}",
                    this.SelectSql, this._IdAttributeName, keyFieldName,
                    tmpTableName, partitionAttrName);
                cmd.CommandType = CommandType.Text;
                cmd.CommandText = sql;
                cmd.CommandTimeout = 60*15;
                IDataReader reader = cmd.ExecuteReader();
                List<int> currentPartitionedEntityIDs = new List<int>();
                List<int> nullPartitionedEntityIDs = new List<int>();
                string lastAttrValue=string.Empty;
                while (reader.Read())
                {
                    int pk = (int)reader[this._MetaData.IdAttributeName];
                    if(reader[partitionAttrName]==null || reader[partitionAttrName]==DBNull.Value)
                    {
                        nullPartitionedEntityIDs.Add(pk);
                    }
                    else
                    {
                        string val = reader[partitionAttrName].ToString();
                        if(val !=lastAttrValue)
                        {
                            int partitionID = partitionedEntityIDs.Count + 1;
                            partitionedEntityIDs.Add(partitionID, currentPartitionedEntityIDs);
                            currentPartitionedEntityIDs = new List<int>();
                            currentPartitionedEntityIDs.Add(pk);
                            lastAttrValue = val;
                        }
                        else
                        {
                            currentPartitionedEntityIDs.Add(pk);
                        }
                    }
                }
                reader.Close();
                if(currentPartitionedEntityIDs.Count>0)
                {
                    int partitionID = partitionedEntityIDs.Count + 1;
                    partitionedEntityIDs.Add(partitionID, currentPartitionedEntityIDs);
                }
                if(nullPartitionedEntityIDs.Count>0)
                {
                    int partitionID = partitionedEntityIDs.Count + 1;
                    partitionedEntityIDs.Add(partitionID, nullPartitionedEntityIDs);
                }
            }
            catch(Exception ex)
            {
                throw new Exception("Unable to read entities from db: " + ex.ToString());
            }
            finally
            {
                conn.Close();
            }
            return partitionedEntityIDs;
        }

        public Range<int> GetRange(string attrName)
        {
            if (this.Source == SourceType.File)
                throw new NotImplementedException("Not implemented");

            Range<int> range = new Range<int>();

            IDbConnection conn = null;
            if (this._DbProvider == DataProviderType.MSSQL)
                conn = new SqlConnection(this._ConnectionString);
            else
                conn = new OracleConnection(this._ConnectionString);
            try
            {
                conn.Open();
                IDbCommand cmd = conn.CreateCommand();
                cmd.CommandType = CommandType.Text;
                cmd.CommandText =
                    string.Format("select max ({0}) from ({1}) temp",
                                  attrName, this._SelectSql);
                IDataReader reader = cmd.ExecuteReader();
                if(reader.Read())
                {
                    range.ToValue = (int) (Math.Ceiling(double.Parse(reader[0].ToString())));
                }
                reader.Close();

                cmd.CommandText =
                    string.Format("select min ({0}) from ({1}) temp",
                                  attrName, this._SelectSql);
                reader = cmd.ExecuteReader();
                if (reader.Read())
                {
                    range.FromValue = (int)(Math.Floor(double.Parse(reader[0].ToString())));
                }
                reader.Close();
            }
            catch (Exception ex)
            {
                throw new Exception("Unable to read entities from db: " + ex.ToString());
            }
            finally
            {
                conn.Close();
            }

            return range;
        }
        #endregion

        public int GetDistinctCount(string attrName)
        {
            if (this.Source == SourceType.File)
                throw new NotImplementedException("Not implemented");

            int distinctCount = 1;

            IDbConnection conn = null;
            if (this._DbProvider == DataProviderType.MSSQL)
                conn = new SqlConnection(this._ConnectionString);
            else
                conn = new OracleConnection(this._ConnectionString);
            try
            {
                conn.Open();
                IDbCommand cmd = conn.CreateCommand();
                cmd.CommandType = CommandType.Text;
                cmd.CommandText =
                    string.Format("select count (distinct {0}) from ({1}) temp",
                                  attrName, this._SelectSql);
                IDataReader reader = cmd.ExecuteReader();
                if (reader.Read())
                {
                    distinctCount = (int)(Math.Ceiling(double.Parse(reader[0].ToString())));
                }
                reader.Close();
            }
            catch (Exception ex)
            {
                throw new Exception("Unable to read entities from db: " + ex.ToString());
            }
            finally
            {
                conn.Close();
            }

            return distinctCount;
        }
    }
}
